package com.bana.sys.socket;

import com.bana.sys.code.SendResult;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import java.net.InetSocketAddress;
import java.util.concurrent.CopyOnWriteArraySet;

/**
 * tcp服务端模式的网络连接
 */
@ChannelHandler.Sharable
public class ServerChannel extends SocketChannel {

    private static Logger logger= LogManager.getLogger(ServerChannel.class);

    /**
     * 客户端集合
     */
    private CopyOnWriteArraySet<ChannelHandlerContext> Clients = new CopyOnWriteArraySet<>();

    /**
     * 监听端口
     */
    private int port;

    /**
     * 构造函数
     * @param port 集群节点端口
     */
    public ServerChannel(int port)
    {
        this.port=port;
    }

    /**
     * 开始监听端口
     */
    public void listen() {
        NioEventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap boot = new ServerBootstrap();
            boot.group(workerGroup);
            boot.channel(NioServerSocketChannel.class);
            boot.localAddress(new InetSocketAddress(port));
            boot.childHandler(this);
            boot.bind().sync();
            logger.info("listen "+port);
        } catch (Exception e){
            workerGroup.shutdownGracefully();
            logger.info(e);
        }
    }

    /**
     * 发送协议
     * @param id 协议编号
     * @param timeStamp 时间戳
     * @param buffer 发送内容
     * @param timeout 等待响应超时时间
     * @return 发送结果
     */
    public SocketResult sendTcp( int id,long timeStamp, byte[] buffer, int timeout)
    {
        //异步回调
        NoticeHandler handler=new NoticeHandler(id,timeStamp,0,0,0);


        //注册异步等待
        addAsyncHandler(handler);
        for(ChannelHandlerContext client:Clients)
        {
            //发送数据
            SendResult result= sendTcp(client,buffer);
            SocketResult sr=new SocketResult();
            sr.setResult(result);
            if(result==SendResult.Success)
            {
                //等待响应
                if (handler.await(timeout))
                {
                    sr.setReceive(handler.getBuffer());
                }
                else
                {
                    sr.setResult(SendResult.Timeout);
                }
            }
            return sr;
        }
        SocketResult sr=new SocketResult();
        sr.setResult(SendResult.NotFoundSocket);
        return sr;
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg){

        receive(ctx,(ByteBuf) msg);
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) {

        Clients.add(ctx);
        logger.info("accept "+ctx.toString());
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) {

        Clients.remove(ctx);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause){

        Clients.remove(ctx);
        logger.info("exception "+ctx.name()+" "+cause.getMessage());
    }
}
